# kafka_consumer.py
from confluent_kafka import Consumer, KafkaError, KafkaException
from utils.logger import logger
import config
import socket
import signal
import sys

class KafkaConsumerWrapper:
    def __init__(self, message_handler):
        self.running = True  # self.running 控制主循环是否继续（便于优雅退出）
        self.message_handler = message_handler  # 是一个外部传进来的回调函数，用来处理每条消息

        self.consumer = Consumer({
            'bootstrap.servers': config.KAFKA_BROKER,
            'client.id': config.Consumer_ID or socket.gethostname(),  # 	当前消费者的唯一标识（IP 地址或自定义）
            'group.id': config.Group_ID,   # 	消费组 ID，决定负载均衡策略
            # 当消费者 首次加入某个 group，且这个 group 没有已提交的 offset 时，Kafka 就会根据这个参数来决定“从哪开始消费”。这个参数只在没有 offset 的时候起作用（即新 group 或 topic）。如果这个 group.id 之前消费过这个 topic，Kafka 会从上次的 offset 开始，无视这个参数。
            # earliest:从分区最早的数据开始消费（从头开始）
            # latest:从分区最新的数据开始消费（跳过历史，只收以后新来的）
            'auto.offset.reset': 'earliest',   # 生产可改为 latest
            # 是否自动提交 offset（这里手动控制）
            'enable.auto.commit': False,       # 手动提交
        })

        logger.info("✅ Kafka Consumer 初始化完成")



    def consume(self, topics):
        """
        主消费循环
        """
        # 调用 subscribe() 订阅你传进来的 topic 列表（例如：["my_topic"]）
        self.consumer.subscribe(topics)
        logger.info(f"🎧 Subscribed to topics: {topics}")

        try:
            while self.running:
                # 每隔 1 秒轮询一次 Kafka，看是否有新消息
                msg = self.consumer.poll(timeout=1.0)
                if msg is None:
                    continue

                if msg.error():
                    #   # 分区结尾，不是错误，跳过
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        logger.debug(f"💤 Reached end of {msg.topic()}[{msg.partition()}] at offset {msg.offset()}")
                    else:
                        logger.error(f"❗ Kafka error: {msg.error()}")
                        raise KafkaException(msg.error())
                else:
                    # 正常消息时，调用你传入的 handle_message(msg)
                    self.message_handler(msg)
                    # 消息处理完后，手动提交 offset，确保“消费-提交一致性”
                    self.consumer.commit(msg)  # 手动提交 offset
        # 支持 Ctrl+C 中断退出
        except KeyboardInterrupt:
            logger.warning("🛑 Received KeyboardInterrupt. Shutting down consumer...")
        # 其他异常统一 log 输出，方便排查
        except Exception as e:
            logger.exception(f"❗ Unexpected error in consumer: {e}")
        finally:
            self.shutdown()

    def shutdown(self):
        """
        优雅关闭
        """
        logger.info("🔻 Closing Kafka consumer...")
        self.running = False
        self.consumer.close()
        logger.info("✅ Kafka consumer closed cleanly")
